/**
 * Alipay.com Inc.
 * Copyright (c) 2004-2020 All Rights Reserved.
 */
package com.dal.flink.window;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.sql.Timestamp;

/**
 * @author dingaolin
 * @version : HopWindowJob.java, v 0.1 2020年10月22日 下午7:56 dingaolin Exp $
 */
public class HopWindowJob {

    public static void main(String[] args) throws Exception {
//        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
//        executionEnvironment.setParallelism(1);
//        executionEnvironment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
//        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(executionEnvironment);
//
//        // mock data
//        DataStream<String> logTextStream = executionEnvironment.socketTextStream("localhost", 9001);
//        DataStream<StationLog> logDataStream = logTextStream.map((MapFunction<String, StationLog>) line -> {
//            String[] split = line.split(",");
//            return new StationLog(split[0], split[1], split[2], Long.valueOf(split[3]));
//        }).assignTimestampsAndWatermarks(WatermarkStrategy.<StationLog>forBoundedOutOfOrderness(Duration.ofSeconds(5))
//                .withTimestampAssigner(new SerializableTimestampAssigner<StationLog>() {
//                    @Override
//                    public long extractTimestamp(StationLog element, long recordTimestamp) {
//                        return element.getCallTime();
//                    }
//                })
//        );
//
//        Table log = tableEnv.fromDataStream(logDataStream, "sid,callIn,callOut,callTime");
//        tableEnv.createTemporaryView("log", log);
//
//        Table teacherResult = tableEnv.sqlQuery("SELECT sid,count(callIn),count(callOut) FROM log group by sid");
//        DataStream<Row> logSinkStream = tableEnv.toAppendStream(teacherResult, Row.class);
//        logSinkStream.print();
//
//        // execute
//        executionEnvironment.execute("HopWindowJob");

        // 获取 environment
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 指定系统时间概念为 event time
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);


        // 初始数据
        DataStream<Tuple3<Long, String, Integer>> logWithTime = env.socketTextStream("localhost", 9001)
                .map(new MapFunction<String, Tuple3<Long, String, Integer>>() {
                    @Override
                    public Tuple3<Long, String, Integer> map(String value) throws Exception {
                        String[] content = value.split(",");
                        return new Tuple3<>(Long.valueOf(content[0]), content[1], Integer.valueOf(content[2]));
                    }})
                .assignTimestampsAndWatermarks(WatermarkStrategy.<Tuple3<Long, String, Integer>>forMonotonousTimestamps()
                .withTimestampAssigner(new SerializableTimestampAssigner<Tuple3<Long, String, Integer>>() {
                    @Override
                    public long extractTimestamp(Tuple3<Long, String, Integer> element, long recordTimestamp) {
                        return element.f0;
                    }

                }));


//        DataStream<Tuple3<Long, String, Integer>> log = env.fromCollection(Arrays.asList(
//                //时间 06:53:00
//                new Tuple3<>(1572591180_000L, "xiao_ming", 300),
//                //时间 06:53:09
//                new Tuple3<>(1572591189_000L, "zhang_san", 303),
//                //时间 06:53:12
//                new Tuple3<>(1572591192_000L, "xiao_li", 204),
//                //时间 06:53:18
//                new Tuple3<>(1572591198_000L, "xiao_bai", 120),
//                //时间 06:53:21
//                new Tuple3<>(1572591201_000L, "li_si", 208)
//        ));

        // 指定时间戳
//        SingleOutputStreamOperator<Tuple3<Long, String, Integer>> logWithTime = log.assignTimestampsAndWatermarks(new AscendingTimestampExtractor<Tuple3<Long, String, Integer>>() {
//
//            @Override
//            public long extractAscendingTimestamp(Tuple3<Long, String, Integer> element) {
//                return element.f0;
//            }
//        });

        // 转换为 Table
        Table logT = tEnv.fromDataStream(logWithTime, "t.rowtime, name, v");

        // HOP(time_attr, interval1, interval2)
        // interval1 滑动长度
        // interval2 窗口长度
        Table result = tEnv.sqlQuery("SELECT HOP_START(t, INTERVAL '5' SECOND, INTERVAL '10' SECOND) AS window_start," +
                "HOP_END(t, INTERVAL '5' SECOND, INTERVAL '10' SECOND) AS window_end, SUM(v) FROM "
                + logT + " GROUP BY HOP(t, INTERVAL '5' SECOND, INTERVAL '10' SECOND)");

        TypeInformation<Tuple3<Timestamp, Timestamp, Integer>> typeInfo = new TypeHint<Tuple3<Timestamp, Timestamp, Integer>>() {
        }.getTypeInfo();
        tEnv.toAppendStream(result, typeInfo).print();

        env.execute("HopWindowJob");
    }


    public static class StationLog {
        public String sid;
        public String callIn;
        public String callOut;
        public Long callTime;

        // 需要无参数构造方法
        public StationLog() {
        }

        public StationLog(String sid, String callIn, String callOut, Long callTime) {
            this.sid = sid;
            this.callIn = callIn;
            this.callOut = callOut;
            this.callTime = callTime;
        }

        /**
         * Getter method for property <tt>sid</tt>.
         *
         * @return property value of sid
         */
        public String getSid() {
            return sid;
        }

        /**
         * Setter method for property <tt>sid</tt>.
         *
         * @param sid value to be assigned to property sid
         */
        public void setSid(String sid) {
            this.sid = sid;
        }

        /**
         * Getter method for property <tt>callIn</tt>.
         *
         * @return property value of callIn
         */
        public String getCallIn() {
            return callIn;
        }

        /**
         * Setter method for property <tt>callIn</tt>.
         *
         * @param callIn value to be assigned to property callIn
         */
        public void setCallIn(String callIn) {
            this.callIn = callIn;
        }

        /**
         * Getter method for property <tt>callOut</tt>.
         *
         * @return property value of callOut
         */
        public String getCallOut() {
            return callOut;
        }

        /**
         * Setter method for property <tt>callOut</tt>.
         *
         * @param callOut value to be assigned to property callOut
         */
        public void setCallOut(String callOut) {
            this.callOut = callOut;
        }

        /**
         * Getter method for property <tt>callTime</tt>.
         *
         * @return property value of callTime
         */
        public Long getCallTime() {
            return callTime;
        }

        /**
         * Setter method for property <tt>callTime</tt>.
         *
         * @param callTime value to be assigned to property callTime
         */
        public void setCallTime(Long callTime) {
            this.callTime = callTime;
        }
    }
}